package com.alison.datastream.chapter6_timeAndWaterMark;


import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @Author alison
 * @Date 2024/4/4 11:10
 * @Version 1.0
 * @Description
 */
public class AssignWatermark {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger(RestOptions.PORT, 8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        //1.12 中默认实现, 过期了
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.getConfig().setAutoWatermarkInterval(1000L);// 每个周期1s生成水印
        DataStreamSource<String> source = env.fromElements(
                "add 1712202113448",
                "sub 1712202114448",
                "flink 1712202115448",
                "add 1712202116448",
                "java 1712202117448",
                "netty 1712202118448"
        );
        DataStream<Tuple3<String, Long, Long>> input = source.map(line -> {
            String[] arr = line.split(" ");
            String id = arr[0];
            long time = Long.parseLong(arr[1]);
            return Tuple3.of(id, 1L, time);
        }).returns(Types.TUPLE(Types.STRING, Types.LONG, Types.LONG));
        DataStream<Tuple3<String, Long, Long>> watermarks = input.assignTimestampsAndWatermarks(
                WatermarkStrategy.forGenerator((context -> new MyPeriodicGenerator()))
                        .withTimestampAssigner((event, recordTimestamp) -> event.f2));
//        使用内置的水印生成器
//        DataStream<Tuple3<String, Long, Long>> watermarks = input.assignTimestampsAndWatermarks(
//                WatermarkStrategy.<Tuple3<String, Long, Long>>forMonotonousTimestamps()
//                .withTimestampAssigner((event, recordTimestamp) -> event.f2));

        DataStream<Tuple3<String, Long, Long>> wordcount = watermarks.keyBy(0)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .sum(1);

        wordcount.print();
        env.execute("watermarks");

    }


    // 定期生成Watermark
    // 数据流元素 Tuple2<String, Long, Long> 共三个字段
    // 第一个字段为单词
    // 第二个字段为词频
    // 第三个字段是时间戳
    public static class MyPeriodicGenerator implements WatermarkGenerator<Tuple3<String, Long, Long>> {

        private final long maxOutOfOrderness = 5 * 1000; // 5 秒钟
        private long currentMaxTimestamp;                 // 已抽取的Timestamp最大值

        // 每个元素都会调用这个方法，如果我们想依赖每个元素生成一个水印，然后发射到下游(可选，就是看是否用output来收集水印)，我们可以实现这个方法.
        @Override
        public void onEvent(Tuple3<String, Long, Long> event, long eventTimestamp, WatermarkOutput output) {
            // 更新currentMaxTimestamp为当前遇到的最大值
            currentMaxTimestamp = Math.max(currentMaxTimestamp, eventTimestamp);
        }

        //  如果数据量比较大的时候，我们每条数据都生成一个水印的话，会影响性能，所以这里还有一个周期性生成水印的方法。这个水印的生成周期可以这样设置：
        @Override
        public void onPeriodicEmit(WatermarkOutput output) {
            // Watermark比currentMaxTimestamp最大值慢
            output.emitWatermark(new Watermark(currentMaxTimestamp - maxOutOfOrderness));
        }

    }

    // 逐个检查数据流中的元素，根据元素中的特殊字段，判断是否要生成Watermark
    // 数据流元素 Tuple3<String, Long, Boolean> 共三个字段
    // 第一个字段为数据本身
    // 第二个字段是时间戳
    // 第三个字段判断是否为Watermark的标记
    public static class MyPunctuatedGenerator implements WatermarkGenerator<Tuple3<String, Long, Boolean>> {

        @Override
        public void onEvent(Tuple3<String, Long, Boolean> event, long eventTimestamp, WatermarkOutput output) {
            if (event.f2) {
                output.emitWatermark(new Watermark(event.f1));
            }
        }

        @Override
        public void onPeriodicEmit(WatermarkOutput output) {
            // 这里不需要做任何事情，因为我们在 onEvent() 方法中生成了Watermark
        }

    }
}
